#ifndef HASHDBOPERATION_H_JM_20130314
#define HASHDBOPERATION_H_JM_20130314

#include "../RootBox.h"
#include "../BoxBaseTypes.h"
#include "../BufferedOutputBox.h"
#include "../../common_bbx/sharedBoxData/HashTableHolder.h"

#include "../../LogConf.h"

// ## jmlib
#include <jmlib/tables/hashTable/HashTable.h>
#include <jmlib/tables/hashTable/BoostHashWrapper.h>

namespace jm4bbx {

	struct HashDBOperationBase {
		static size_t HASH_TABLE_SIZE;
		static const char* INPUT_COUNT_PARAM_NAME;
	};


	template<typename T, class ActivityHandler>
	class  HashDBOperation : public HashDBOperationBase, 
		public BufferedOneOutputBox<BinaryOperationResultIndexType>  {
			JM_USE_LOGGER(HashDBOperationLogger)
			JM_USE_EXPLICIT_LOGGER(ANALYSE_HT_LOGGER)
	public:
		
		static const  groupType OUTPUT_GROUP;	
		static const  columnType OUTPUT_COLUMN;
		static const  outType OUT_ARC;

		typedef T KeyType;
		typedef KeyMultiRecords<T> ValueType;
		typedef KeyRecord<T> TransferValueType;
		typedef typename HashTableHolder<T>::HashTable HashTable;

		

		HashDBOperation(const BoxParamPack& p) 
			: BufferedOneOutputBox(p, OUTPUT_GROUP, OUTPUT_COLUMN, OUT_ARC),
			hashTable_(HASH_TABLE_SIZE, HashTable::HashFunctorPtr(
			new jmlib::BoostHashWrapper<KeyType>())),
			handler_(this, p) {

				workDone_ = false;

				inputs_ = 
					getBoxParameterValue(p, INPUT_COUNT_PARAM_NAME, 1);
				// na 0 je vystup takze vstupy od 1
				actualInput_ = 1;

				//double f = 1/(double)inputs_;
				//hashTable_.realloc(3000000 * f);
				hashTable_.setSynchronized(false);
		}

		virtual ~HashDBOperation() {
			hashTable_.clear();
		}

		virtual void initImpl() override {

			add_group(OUTPUT_GROUP, OUTPUT_COLUMN, 
				get_output_descriptors(OUT_ARC));

			for (int i = 1; i <= inputs_; ++i) {
				// Vstupy dat prveho operandu
				add_group(groupType(i), columnType(i),
					get_input_descriptors(inType(i - 1)));
				// Vstupy dat druheho operandu
				add_group(groupType(i + inputs_), columnType(i + inputs_),
					get_input_descriptors(inType(i + inputs_ - 1)));
			}


			initBufferedOutputs();
			wait4SignalsIn();
			wait4Receive();
//			receive(IN_ARC_2, INPUT_GROUP_2);
		}
		
		void wait4Receive() {
			receive(inType(actualInput_-1), groupType(actualInput_));
		}
		
		GroupPtr getActualGroup() {
			if (!actualGroup_) {
				actualGroup_ = getGroupPtr(groupType(actualInput_));
			}
			return actualGroup_;
		}

		/// Posun sa na spracovavanie dalsieho vstupu dat.
		/// @return true - presun uspesny a spracovavanie moze pokracovat
		/// @return false - vsetky vstupy spracovane presun sa teda nepodaril
		bool activateNextInput() {
			++actualInput_;
			// 2*pocet vstupov pretoze vstupy su pre oba operandy individualne
			// +1 pretoze na 0 je vystup
			if (actualInput_ < (2*inputs_+1)) {
				actualGroup_ = getGroupPtr(groupType(actualInput_));
				return true;
			}
			else {
				return false;
			}
		}

		bool isFirstStep() {
			// na 0 je vystup -> +1
			// Prvych inputs_ vstupov patri prvemu kroku a
			// zvysnych inputs_ krokov druhemu kroku.
			return actualInput_ < inputs_ + 1;
		}
		
		bool isLastInput() {
			// na 0 je vystup
			return actualInput_ == (2*inputs_);
		}


		void analyseHashTable() {
			size_t sumCount = 0;
			hashTable_.foreachExtended(
				[&sumCount](HashTable::ExtendedHashEntry& entry) {						
					sumCount += entry.plainEntry.value.size();
			});
			double alf = sumCount / (double) hashTable_.getSize();
			double lf = hashTable_.getLoadFactor();
			LOG_I_EX(ANALYSE_HT_LOGGER, timeReportPrefix << " Aggregated factor == " << alf << "  Plain load factor == " << lf);
		}

		virtual void syncMachEtwas() override {	
			
			GroupPtr ig = getGroupPtr(groupType(actualInput_));//getActualGroup();
			if (ig->isPoisoned() || workDone_) {
				LOG_D("Is poisoned.");
				if (!activateNextInput()) {
					LOG_D("No more input.");
					if (!isOutputEmpty()) {
						LOG_D("Output is not empty -> sending data.");
						workDone_ = true;
						sendOutputData();
						return;
					}
					else {
						LOG_D("Output is empty -> END.");
						actualInput_ = 1;
						send_poisoned(OUT_ARC); // Aby ciel vedel ze je hotovo!!!
						sendSignalsOut(true);
					//	analyseHashTable();
					}
				}
	
				LOG_D("Wait ... & return.");
				wait4Receive();
				return;
			}
			if (isFirstStep()) {
				LOG_D("Do the first step.");
				//int counter = 0;
				
				for (; ig->getReadPosition() < ig->getReadEnd(); ig->moveReadPosition()) {
					//++counter;
					TransferValueType value = 
						get_data<TransferValueType>(columnType(actualInput_))[ig->getReadPosition()];		
					
					auto key = value.getKey();

					if (!hashTable_.addIf(key, ValueType(value.getIndex(), key), 
						jmlib::HashRowActionCondition::HAS_NOT_KEY)) {						
							hashTable_.applyOnExtended(key, 
								[&](HashTable::ExtendedHashEntry& entry){
									// Pri rovnosti klucov sa nevklada do mapy ale sa 
									// len prida index -> pocitadlo zaznamov s rovnakym 
									// klucom musi byt vzdy 1
									//if (entry.getCount() != 1) {
									//	LOG_E("entry.count != 1 ale entry.count == " 
									//		<< entry.getCount());
									//}
									assert(1 == entry.getCount());							
									entry.plainEntry.value.insert(value.getIndex());
							});
					}
				}
				//LOG_D("Hashed " << counter << " records.");
				//LOG_D("Wait ... & end.");
				wait4Receive();
			}
			else {
				LOG_D("Do the second step.");
				//int counter = 0;
				for (; ig->getReadPosition() < ig->getReadEnd(); ig->moveReadPosition()) {
					//++counter;
					TransferValueType value = 
						get_data<TransferValueType>(columnType(actualInput_))[ig->getReadPosition()];		

					auto key = value.getKey();

					if (!hashTable_.applyOnExtended(key, 
						[&](HashTable::ExtendedHashEntry& entry) {
							assert(entry.getCount() == 1);
							handler_.handleRecord(value, entry);
						})) {
							handler_.handleNoRecord(value);
					}
				}		
		
				//LOG_D("Joined " << counter << " records.");

				if (isOutputFull()) {
					LOG_D("Output buffer is full -> sending data.") ;
					sendOutputData();			
				}
				//LOG_D("Wait ... & end.");
				wait4Receive();
			}
		}
	

	void handleRecord(TransferValueType value, typename HashTable::ExtendedHashEntry& entry) {

				assert(entry.getCount() == 1);

				auto end = entry.plainEntry.value.end();
				for (auto it = entry.plainEntry.value.begin(); it != end; ++it) {				
					BinaryOperationResultIndex index = {*it, value.getIndex()};

					addOutputData(index);
				}	
		} 

	private:
		HashTable hashTable_;
		int inputs_;
		int actualInput_;
		GroupPtr actualGroup_;
		bool workDone_;
		ActivityHandler handler_;
	};

	template<typename T, class H>
	const  groupType jm4bbx::HashDBOperation<T, H>::OUTPUT_GROUP = groupType(0);

	template<typename T, class H>
	const  columnType jm4bbx::HashDBOperation<T, H>::OUTPUT_COLUMN = columnType(0);

	template<typename T, class H>
	const  outType jm4bbx::HashDBOperation<T, H>::OUT_ARC = outType(0);

} // namespace jm4bbx


#endif
